package com.sh.data.engine.domain.datadev.flink.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.DateUtil;
import com.sh.data.engine.common.util.FlinkTaskScheduleCycleUtil;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkNodeConfigDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkScheduleConfigDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkScheduleDayRangeDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkTaskDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkTaskService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * @author：jingchen.hr
 * @descripion:
 * @data:6/14/20225:40 PM
 */
@Service
public class FlinkTaskServiceImpl implements FlinkTaskService {

    @Autowired
    private FlinkService flinkService;

    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;

    private static final String ITEM_ENV = "env";
    private static final String ITEM_SCHEDULE = "schedule";

    @Override
    public void delete(Long nodeId) {
        flinkNodeConfigService.deleteByNodeId(nodeId);
    }

    @Override
    public FlinkTaskDomain getActiveByNodeId(Long nodeId) {

        FlinkNodeConfigEntity activeByNodeId = flinkNodeConfigService.getActiveByNodeId(nodeId);
        if (Objects.isNull(activeByNodeId)) {
            return null;
        }
        FlinkTaskDomain flinkTaskDomain = new FlinkTaskDomain();
        flinkTaskDomain.init(activeByNodeId);
        return flinkTaskDomain;
    }

    @Override
    public FlinkTaskDomain getById(Long id) {
        final FlinkNodeConfigDomain nodeConfigById = flinkNodeConfigService.getNodeConfigById(id);
        return ConvertUtil.copyProperties(nodeConfigById, FlinkTaskDomain.class);
    }

    @Override
    public List<FlinkTaskDomain> getAllRunnableTaskList() {
        final List<FlinkNodeConfigEntity> runnableTaskList =
            flinkNodeConfigService.getRunnableTaskList(FlinkNodeTypeEnum.TASK_JAR);
        ArrayList<FlinkTaskDomain> list = Lists.newArrayList();
        runnableTaskList.forEach(
            i -> {
                FlinkTaskDomain flinkTaskDomain = new FlinkTaskDomain();
                flinkTaskDomain.init(i);
                list.add(flinkTaskDomain);
            });
        return list;
    }

    @Override
    public List<FlinkTaskDomain> getAllTaskList() {
        final List<FlinkNodeConfigEntity> allTaskList =
            flinkNodeConfigService.getAllTaskList(FlinkNodeTypeEnum.TASK_JAR);
        ArrayList<FlinkTaskDomain> list = Lists.newArrayList();
        allTaskList.forEach(
            i -> {
                FlinkTaskDomain flinkTaskDomain = new FlinkTaskDomain();
                flinkTaskDomain.init(i);
                list.add(flinkTaskDomain);
            });
        return list;
    }

    @Override
    public List<FlinkTaskDomain> getAllStoppableTaskList() {
        List<FlinkNodeConfigEntity> stopedTaskList =
            flinkNodeConfigService.getStopedTaskList(FlinkNodeTypeEnum.TASK_JAR);
        ArrayList<FlinkTaskDomain> list = Lists.newArrayList();
        stopedTaskList.forEach(
            i -> {
                FlinkTaskDomain flinkTaskDomain = new FlinkTaskDomain();
                flinkTaskDomain.init(i);
                list.add(flinkTaskDomain);
            });
        return list;
    }

    @Override
    public void publish(Long nodeId, Integer status, String userId, String remark) {
        final FlinkTaskDomain flinkTaskDomain = this.getActiveByNodeId(nodeId);
        if (null == flinkTaskDomain) {
            throw new BusinessException("数据不存在");
        }

        if (Objects.equals(flinkTaskDomain.getIsPublished(), status)) {
            return;
        }
        FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        flinkNodeConfigEntity.setId(flinkTaskDomain.getId());
        flinkNodeConfigEntity.setIsPublished(status);
        flinkNodeConfigEntity.setIsActive(1);
        if (Objects.equals(status, 1)) {
            flinkNodeConfigEntity.setPublishTime(new Date());
            flinkNodeConfigEntity.setPublisherId(userId);
        } else {
            flinkNodeConfigEntity.setPublishTime(null);
            flinkNodeConfigEntity.setPublisherId(null);
        }
        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper
            .lambda()
            .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
            .eq(FlinkNodeConfigEntity::getIsActive, 1);

        flinkNodeConfigService.update(flinkNodeConfigEntity, wrapper);
    }

    @Override
    public void rollbackVersion(Long nodeId, Long versionId) {

        flinkNodeConfigService.rollbackVersion(FlinkNodeTypeEnum.TASK_JAR, nodeId, versionId);
    }

    @Override
    public void saveBaseConfig(
        Long nodeId,
        Long resourceNodeId,
        String mainClass,
        String remark,
        String userId,
        Long projectId) {

        final FlinkTaskDomain flinkTaskDomain = getActiveByNodeId(nodeId);

        // FlinkTaskEntity entity = new FlinkTaskEntity();
        FlinkNodeConfigEntity entity = new FlinkNodeConfigEntity();
        // 如果是已发布状态，先把当前这个版本的isActive设为0
        if (null != flinkTaskDomain && Objects.equals(1, flinkTaskDomain.getIsPublished())) {
            FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
            flinkNodeConfigEntity.setId(flinkTaskDomain.getId());
            flinkNodeConfigEntity.setIsActive(0);
            flinkNodeConfigEntity.setUpdateTime(new Date());
            flinkNodeConfigService.updateById(flinkNodeConfigEntity);

            entity = ConvertUtil.copyProperties(flinkTaskDomain, FlinkNodeConfigEntity.class);
            entity.setId(null);
        }
        FlinkNodeConfigEntity.JarTaskConfig jarTaskConfig = new FlinkNodeConfigEntity.JarTaskConfig();
        if (null == flinkTaskDomain || Objects.equals(1, flinkTaskDomain.getIsPublished())) {

            entity.setProjectId(projectId);
            entity.setNodeId(nodeId);
            jarTaskConfig.setResourceNodeId(resourceNodeId);
            jarTaskConfig.setMainClass(mainClass);
            entity.setIsPublished(0);
            entity.setCreatorId(userId);
            entity.setUpdaterId(userId);
            entity.setRemark(remark);
            entity.setIsActive(1);
            entity.setCreateTime(new Date());
            entity.setUpdateTime(new Date());
            entity.setJarTaskConfig(jarTaskConfig);
            if (null == flinkTaskDomain) {
                entity.setEnvConfig(flinkService.getDefaultEnv());
            }
            flinkNodeConfigService.save(entity);
            return;
        }

        entity.setId(flinkTaskDomain.getId());
        jarTaskConfig.setResourceNodeId(resourceNodeId);
        jarTaskConfig.setMainClass(mainClass);
        entity.setRemark(remark);
        entity.setUpdaterId(userId);
        entity.setUpdateTime(new Date());
        entity.setJarTaskConfig(jarTaskConfig);
        flinkNodeConfigService.updateById(entity);
    }

    @Override
    public void saveEnv(Long nodeId, String envConfig, String userId, Long projectId) {
        saveConfig(nodeId, userId, projectId, envConfig, ITEM_ENV);
    }

    @Override
    public void saveSchedule(Long nodeId, String scheduleConfig, String userId, Long projectId) {
        saveConfig(nodeId, userId, projectId, scheduleConfig, ITEM_SCHEDULE);
    }

    private void saveConfig(
        Long nodeId, String userId, Long projectId, String content, String item) {
        final FlinkTaskDomain flinkTaskDomain = getActiveByNodeId(nodeId);

        if (null != flinkTaskDomain && StringUtils.isNotBlank(content)) {
            if (ITEM_ENV.equals(item) && content.equals(flinkTaskDomain.getEnvConfig())) {
                return;
            } else if (ITEM_SCHEDULE.equals(item)
                && content.equals(flinkTaskDomain.getScheduleConfig())) {
                return;
            }
        }
        FlinkNodeConfigEntity entity = new FlinkNodeConfigEntity();
        // 如果是已发布状态，先把当前这个版本的isActive设为0
        if (null != flinkTaskDomain && Objects.equals(1, flinkTaskDomain.getIsPublished())) {
            FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
            flinkNodeConfigEntity.setId(flinkTaskDomain.getId());
            flinkNodeConfigEntity.setIsActive(0);
            flinkNodeConfigEntity.setUpdateTime(new Date());
            flinkNodeConfigService.updateById(flinkNodeConfigEntity);

            entity = ConvertUtil.copyProperties(flinkTaskDomain, FlinkNodeConfigEntity.class);
            entity.setId(null);
        }

        if (null == flinkTaskDomain || Objects.equals(1, flinkTaskDomain.getIsPublished())) {
            entity.setProjectId(projectId);
            entity.setNodeId(nodeId);
            entity.setIsPublished(0);
            entity.setCreatorId(userId);
            entity.setUpdaterId(userId);
            entity.setIsActive(1);
            entity.setCreateTime(new Date());
            entity.setUpdateTime(new Date());
            saveConfig(entity, content, item);
            if (null == flinkTaskDomain) {
                entity.setEnvConfig(flinkService.getDefaultEnv());
            }
            flinkNodeConfigService.save(entity);
            return;
        }

        entity.setId(flinkTaskDomain.getId());
        saveConfig(entity, content, item);
        entity.setUpdaterId(userId);
        entity.setUpdateTime(new Date());
        entity.setIsActive(flinkTaskDomain.getIsActive());
        flinkNodeConfigService.updateById(entity);
    }

    private void saveConfig(FlinkNodeConfigEntity entity, String content, String item) {
        if (ITEM_ENV.equals(item)) {
            entity.setEnvConfig(content);
        } else if (ITEM_SCHEDULE.equals(item)) {
            // fixme 验证content是否为ScheduleConfig的json格式
            FlinkNodeConfigEntity.FlinkScheduleConfig scheduleConfig =
                JSON.parseObject(content, FlinkNodeConfigEntity.FlinkScheduleConfig.class);
            entity.setScheduleConfig(scheduleConfig);
        }
    }

    @Override
    public void reCalculateNextStartAndStopTime(Long nodeId, boolean includeToday) {
        final FlinkTaskDomain domain = getActiveByNodeId(nodeId);
        final FlinkScheduleConfigDomain scheduleConfigDomain = domain.getScheduleConfigDomain();
        if (null == scheduleConfigDomain) {
            return;
        }
        final List<FlinkScheduleDayRangeDomain> rangeList = scheduleConfigDomain.getRangeList();

        List<ImmutablePair<Date, Date>> effectiveDateStartEndPairList = Lists.newLinkedList();

        String dateFormat = "yyyy-MM-dd HH:mm:ss";

        for (FlinkScheduleDayRangeDomain flinkScheduleDayRangeDomain : rangeList) {
            effectiveDateStartEndPairList.add(
                new ImmutablePair<>(
                    DateUtil.toDate(
                        flinkScheduleDayRangeDomain.getEffectiveDateStart() + " 00:00:00", dateFormat),
                    DateUtil.toDate(
                        flinkScheduleDayRangeDomain.getEffectiveDateEnd() + " 23:59:59", dateFormat)));
        }

        final Date nextStartTime =
            FlinkTaskScheduleCycleUtil.getNextStartTime(
                scheduleConfigDomain.getEffectDayType(),
                effectiveDateStartEndPairList,
                scheduleConfigDomain.getStartTimeHour(),
                scheduleConfigDomain.getStartTimeMin(),
                includeToday);

        final Date nextStopTime =
            FlinkTaskScheduleCycleUtil.getNextStopTime(
                scheduleConfigDomain.getEffectDayType(),
                effectiveDateStartEndPairList,
                nextStartTime,
                scheduleConfigDomain.getStartTimeHour(),
                scheduleConfigDomain.getStartTimeMin(),
                scheduleConfigDomain.getEndTimeHour(),
                scheduleConfigDomain.getEndTimeMin());

        FlinkNodeConfigEntity entity = new FlinkNodeConfigEntity();

        entity.setId(domain.getId());
        entity.setIsActive(1);
        entity.setNextExecTime(nextStartTime);
        entity.setNextStopTime(nextStopTime);
        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper
            .lambda()
            .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
            .eq(FlinkNodeConfigEntity::getIsActive, 1);
        flinkNodeConfigService.update(
            entity, wrapper);
    }
}
